satellite: use ranged loop with GC-GF peer

Peer for generating bloom filters will be able to use ranged loop.

As an addition some cleanup were made:
* remove unused parts of GC BF peer (identity, version control)
* added missing Close method for ranged loop service
* some additional tests added

https://github.com/storj/storj/issues/5545

Change-Id: I9a3d85f5fffd2ebc7f2bf7ed024220117ab2be29

cmd/satellite/gc-bf.go

@@ -22,12 +22,6 @@ func cmdGCBloomFilterRun(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-gc-bloomfilter"})
 	if err != nil {
 		return errs.New("Error starting master database on satellite GC: %+v", err)
@@ -53,12 +47,7 @@ func cmdGCBloomFilterRun(cmd *cobra.Command, args []string) (err error) {
 		err = errs.Combine(err, revocationDB.Close())
 	}()
 
-	peer, err := satellite.NewGarbageCollectionBF(log, identity, db, metabaseDB, revocationDB, version.Build, &runCfg.Config, process.AtomicLevel(cmd))
-	if err != nil {
-		return err
-	}
-
-	_, err = peer.Version.Service.CheckVersion(ctx)
+	peer, err := satellite.NewGarbageCollectionBF(log, db, metabaseDB, revocationDB, version.Build, &runCfg.Config, process.AtomicLevel(cmd))
 	if err != nil {
 		return err
 	}

private/testplanet/satellite.go

@@ -564,7 +564,7 @@ func (planet *Planet) newSatellite(ctx context.Context, prefix string, index int
 		return nil, errs.Wrap(err)
 	}
 
-	gcBFPeer, err := planet.newGarbageCollectionBF(ctx, index, identity, db, metabaseDB, config, versionInfo)
+	gcBFPeer, err := planet.newGarbageCollectionBF(ctx, index, db, metabaseDB, config, versionInfo)
 	if err != nil {
 		return nil, errs.Wrap(err)
 	}
@@ -763,7 +763,7 @@ func (planet *Planet) newGarbageCollection(ctx context.Context, index int, ident
 	return satellite.NewGarbageCollection(log, identity, db, metabaseDB, revocationDB, versionInfo, &config, nil)
 }
 
-func (planet *Planet) newGarbageCollectionBF(ctx context.Context, index int, identity *identity.FullIdentity, db satellite.DB, metabaseDB *metabase.DB, config satellite.Config, versionInfo version.Info) (_ *satellite.GarbageCollectionBF, err error) {
+func (planet *Planet) newGarbageCollectionBF(ctx context.Context, index int, db satellite.DB, metabaseDB *metabase.DB, config satellite.Config, versionInfo version.Info) (_ *satellite.GarbageCollectionBF, err error) {
 	defer mon.Task()(&ctx)(&err)
 
 	prefix := "satellite-gc-bf" + strconv.Itoa(index)
@@ -774,7 +774,7 @@ func (planet *Planet) newGarbageCollectionBF(ctx context.Context, index int, ide
 		return nil, errs.Wrap(err)
 	}
 	planet.databases = append(planet.databases, revocationDB)
-	return satellite.NewGarbageCollectionBF(log, identity, db, metabaseDB, revocationDB, versionInfo, &config, nil)
+	return satellite.NewGarbageCollectionBF(log, db, metabaseDB, revocationDB, versionInfo, &config, nil)
 }
 
 func (planet *Planet) newRangedLoop(ctx context.Context, index int, db satellite.DB, metabaseDB *metabase.DB, config satellite.Config) (_ *satellite.RangedLoop, err error) {

satellite/gc-bf.go

@@ -14,15 +14,13 @@ import (
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 
-	"storj.io/common/identity"
 	"storj.io/common/peertls/extensions"
-	"storj.io/common/storj"
 	"storj.io/private/debug"
 	"storj.io/private/version"
 	"storj.io/storj/private/lifecycle"
-	version_checker "storj.io/storj/private/version/checker"
 	"storj.io/storj/satellite/gc/bloomfilter"
 	"storj.io/storj/satellite/metabase"
+	"storj.io/storj/satellite/metabase/rangedloop"
 	"storj.io/storj/satellite/metabase/segmentloop"
 	"storj.io/storj/satellite/overlay"
 )
@@ -31,18 +29,12 @@ import (
 //
 // architecture: Peer
 type GarbageCollectionBF struct {
-	Log      *zap.Logger
-	Identity *identity.FullIdentity
-	DB       DB
+	Log *zap.Logger
+	DB  DB
 
 	Servers  *lifecycle.Group
 	Services *lifecycle.Group
 
-	Version struct {
-		Chore   *version_checker.Chore
-		Service *version_checker.Service
-	}
-
 	Debug struct {
 		Listener net.Listener
 		Server   *debug.Server
@@ -60,16 +52,18 @@ type GarbageCollectionBF struct {
 		Config  bloomfilter.Config
 		Service *bloomfilter.Service
 	}
+
+	RangedLoop struct {
+		Service *rangedloop.Service
+	}
 }
 
 // NewGarbageCollectionBF creates a new satellite garbage collection peer which collects storage nodes bloom filters.
-func NewGarbageCollectionBF(log *zap.Logger, full *identity.FullIdentity, db DB,
-	metabaseDB *metabase.DB, revocationDB extensions.RevocationDB,
+func NewGarbageCollectionBF(log *zap.Logger, db DB, metabaseDB *metabase.DB, revocationDB extensions.RevocationDB,
 	versionInfo version.Info, config *Config, atomicLogLevel *zap.AtomicLevel) (*GarbageCollectionBF, error) {
 	peer := &GarbageCollectionBF{
-		Log:      log,
-		Identity: full,
-		DB:       db,
+		Log: log,
+		DB:  db,
 
 		Servers:  lifecycle.NewGroup(log.Named("servers")),
 		Services: lifecycle.NewGroup(log.Named("services")),
@@ -94,45 +88,49 @@ func NewGarbageCollectionBF(log *zap.Logger, full *identity.FullIdentity, db DB,
 		})
 	}
 
-	{ // setup version control
-		peer.Log.Info("Version info",
-			zap.Stringer("Version", versionInfo.Version.Version),
-			zap.String("Commit Hash", versionInfo.CommitHash),
-			zap.Stringer("Build Timestamp", versionInfo.Timestamp),
-			zap.Bool("Release Build", versionInfo.Release),
-		)
-		peer.Version.Service = version_checker.NewService(log.Named("version"), config.Version, versionInfo, "Satellite")
-		peer.Version.Chore = version_checker.NewChore(peer.Version.Service, config.Version.CheckInterval)
-
-		peer.Services.Add(lifecycle.Item{
-			Name: "version",
-			Run:  peer.Version.Chore.Run,
-		})
-	}
-
 	{ // setup overlay
 		peer.Overlay.DB = peer.DB.OverlayCache()
 	}
 
-	{ // setup metainfo
-		peer.Metainfo.SegmentLoop = segmentloop.New(
-			log.Named("segmentloop"),
-			config.Metainfo.SegmentLoop,
-			metabaseDB,
-		)
-		peer.Services.Add(lifecycle.Item{
-			Name:  "metainfo:segmentloop",
-			Run:   peer.Metainfo.SegmentLoop.Run,
-			Close: peer.Metainfo.SegmentLoop.Close,
-		})
-	}
-
 	{ // setup garbage collection bloom filters
 		log := peer.Log.Named("garbage-collection-bf")
 		peer.GarbageCollection.Config = config.GarbageCollectionBF
 		if config.GarbageCollectionBF.UseRangedLoop {
 			log.Info("using ranged loop")
+
+			provider := rangedloop.NewMetabaseRangeSplitter(metabaseDB, config.RangedLoop.AsOfSystemInterval, config.RangedLoop.BatchSize)
+			peer.RangedLoop.Service = rangedloop.NewService(log.Named("rangedloop"), config.RangedLoop, provider, []rangedloop.Observer{
+				bloomfilter.NewObserver(log.Named("gc-bf"),
+					config.GarbageCollectionBF,
+					peer.Overlay.DB,
+				),
+			})
+
+			if !config.GarbageCollectionBF.RunOnce {
+				peer.Services.Add(lifecycle.Item{
+					Name:  "garbage-collection-bf",
+					Run:   peer.RangedLoop.Service.Run,
+					Close: peer.RangedLoop.Service.Close,
+				})
+				peer.Debug.Server.Panel.Add(
+					debug.Cycle("Garbage Collection Bloom Filters", peer.RangedLoop.Service.Loop))
+			}
 		} else {
+			log.Info("using segments loop")
+
+			{ // setup metainfo
+				peer.Metainfo.SegmentLoop = segmentloop.New(
+					log.Named("segmentloop"),
+					config.Metainfo.SegmentLoop,
+					metabaseDB,
+				)
+				peer.Services.Add(lifecycle.Item{
+					Name:  "metainfo:segmentloop",
+					Run:   peer.Metainfo.SegmentLoop.Run,
+					Close: peer.Metainfo.SegmentLoop.Close,
+				})
+			}
+
 			peer.GarbageCollection.Service = bloomfilter.NewService(
 				log,
 				config.GarbageCollectionBF,
@@ -158,6 +156,9 @@ func NewGarbageCollectionBF(log *zap.Logger, full *identity.FullIdentity, db DB,
 func (peer *GarbageCollectionBF) Run(ctx context.Context) (err error) {
 	defer mon.Task()(&ctx)(&err)
 
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	group, ctx := errgroup.WithContext(ctx)
 
 	pprof.Do(ctx, pprof.Labels("subsystem", "gc-bloomfilter"), func(ctx context.Context) {
@@ -165,12 +166,20 @@ func (peer *GarbageCollectionBF) Run(ctx context.Context) (err error) {
 		peer.Services.Run(ctx, group)
 
 		if peer.GarbageCollection.Config.RunOnce {
-			err = peer.GarbageCollection.Service.RunOnce(ctx)
-		} else {
-			pprof.Do(ctx, pprof.Labels("name", "subsystem-wait"), func(ctx context.Context) {
-				err = group.Wait()
+			group.Go(func() error {
+				if peer.GarbageCollection.Config.UseRangedLoop {
+					_, err = peer.RangedLoop.Service.RunOnce(ctx)
+				} else {
+					err = peer.GarbageCollection.Service.RunOnce(ctx)
+				}
+				cancel()
+				return err
 			})
 		}
+
+		pprof.Do(ctx, pprof.Labels("name", "subsystem-wait"), func(ctx context.Context) {
+			err = group.Wait()
+		})
 	})
 
 	return err
@@ -183,6 +192,3 @@ func (peer *GarbageCollectionBF) Close() error {
 		peer.Services.Close(),
 	)
 }
-
-// ID returns the peer ID.
-func (peer *GarbageCollectionBF) ID() storj.NodeID { return peer.Identity.ID }

satellite/gc-bf_test.go

@@ -0,0 +1,45 @@
+// Copyright (C) 2023 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package satellite_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+
+	"storj.io/common/testcontext"
+	"storj.io/storj/private/testplanet"
+	"storj.io/storj/satellite"
+)
+
+func TestGCBFUseRangedLoop(t *testing.T) {
+	testplanet.Run(t, testplanet.Config{
+		SatelliteCount: 1,
+		Reconfigure: testplanet.Reconfigure{
+			Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
+				config.GarbageCollectionBF.RunOnce = true
+				config.GarbageCollectionBF.UseRangedLoop = true
+			},
+		},
+	}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
+		err := planet.Satellites[0].GCBF.Run(ctx)
+		require.NoError(t, err)
+	})
+}
+
+func TestGCBFUseSegmentsLoop(t *testing.T) {
+	testplanet.Run(t, testplanet.Config{
+		SatelliteCount: 1,
+		Reconfigure: testplanet.Reconfigure{
+			Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
+				config.GarbageCollectionBF.RunOnce = true
+				config.GarbageCollectionBF.UseRangedLoop = false
+			},
+		},
+	}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
+		err := planet.Satellites[0].GCBF.Run(ctx)
+		require.NoError(t, err)
+	})
+}

satellite/metabase/rangedloop/service.go

@@ -77,6 +77,12 @@ type ObserverDuration struct {
 	Duration time.Duration
 }
 
+// Close stops the ranged loop.
+func (service *Service) Close() error {
+	service.Loop.Close()
+	return nil
+}
+
 // Run starts the looping service.
 func (service *Service) Run(ctx context.Context) (err error) {
 	defer mon.Task()(&ctx)(&err)

satellite/metabase/rangedloop/service_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strconv"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -14,11 +15,20 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap/zaptest"
 
+	"storj.io/common/memory"
 	"storj.io/common/testcontext"
+	"storj.io/common/testrand"
 	"storj.io/common/uuid"
+	"storj.io/storj/private/testplanet"
+	"storj.io/storj/satellite/accounting/nodetally"
+	"storj.io/storj/satellite/audit"
+	"storj.io/storj/satellite/gc/bloomfilter"
+	"storj.io/storj/satellite/gracefulexit"
 	"storj.io/storj/satellite/metabase/rangedloop"
 	"storj.io/storj/satellite/metabase/rangedloop/rangedlooptest"
 	"storj.io/storj/satellite/metabase/segmentloop"
+	"storj.io/storj/satellite/metrics"
+	"storj.io/storj/satellite/repair/checker"
 )
 
 func TestLoopCount(t *testing.T) {
@@ -359,3 +369,64 @@ func TestLoopContinuesAfterObserverError(t *testing.T) {
 	require.Equal(t, observerDurations[4].Duration, -1*time.Second)
 	require.Equal(t, observerDurations[5].Duration, -1*time.Second)
 }
+
+func TestAllInOne(t *testing.T) {
+	testplanet.Run(t, testplanet.Config{
+		SatelliteCount: 1, StorageNodeCount: 4, UplinkCount: 1,
+	}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
+		log := zaptest.NewLogger(t)
+		satellite := planet.Satellites[0]
+
+		for i := 0; i < 100; i++ {
+			err := planet.Uplinks[0].Upload(ctx, satellite, "testbucket", "object"+strconv.Itoa(i), testrand.Bytes(5*memory.KiB))
+			require.NoError(t, err)
+		}
+
+		require.NoError(t, planet.Uplinks[0].CreateBucket(ctx, satellite, "bf-bucket"))
+
+		metabaseProvider := rangedloop.NewMetabaseRangeSplitter(satellite.Metabase.DB, 0, 10)
+
+		config := rangedloop.Config{
+			Parallelism: 8,
+			BatchSize:   3,
+		}
+
+		bfConfig := satellite.Config.GarbageCollectionBF
+		bfConfig.Bucket = "bf-bucket"
+		accessGrant, err := planet.Uplinks[0].Access[satellite.ID()].Serialize()
+		require.NoError(t, err)
+		bfConfig.AccessGrant = accessGrant
+
+		service := rangedloop.NewService(log, config, metabaseProvider, []rangedloop.Observer{
+			rangedloop.NewLiveCountObserver(),
+			metrics.NewObserver(),
+			nodetally.NewRangedLoopObserver(log.Named("accounting:nodetally"),
+				satellite.DB.StoragenodeAccounting(),
+			),
+			audit.NewObserver(log.Named("audit"),
+				satellite.DB.VerifyQueue(),
+				satellite.Config.Audit,
+			),
+			gracefulexit.NewObserver(log.Named("gracefulexit:observer"),
+				satellite.DB.GracefulExit(),
+				satellite.DB.OverlayCache(),
+				satellite.Config.GracefulExit,
+			),
+			bloomfilter.NewObserver(log.Named("gc-bf"),
+				bfConfig,
+				satellite.DB.OverlayCache(),
+			),
+			checker.NewRangedLoopObserver(
+				log.Named("repair:checker"),
+				satellite.DB.RepairQueue(),
+				satellite.Overlay.Service,
+				satellite.Config.Checker,
+			),
+		})
+
+		for i := 0; i < 5; i++ {
+			_, err = service.RunOnce(ctx)
+			require.NoError(t, err, "iteration %d", i+1)
+		}
+	})
+}