JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite/overlay: add nodeevents.DB to satellite overlay service

Browse Source 
Add nodeevents.DB to satellite overlay service so we can insert node
events into the nodeevents DB.

Change-Id: I642c0ccc9941ecdb08cb22d5c8cf701959a55156
This commit is contained in:
Cameron 2022-10-31 17:33:17 -04:00 committed by Storj Robot
parent d21bbab2b2
commit f06da25c3d
12 changed files with 22 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cmd/satellite/fetchpieces.go
View File

@ -89,6 +89,7 @@ func cmdFetchPieces(cmd *cobra.Command, args []string) (err error) {
db.RepairQueue(),
db.Buckets(),
db.OverlayCache(),
db.NodeEvents(),
db.Reputation(),
db.Containment(),
rollupsWriteCache,

1
cmd/satellite/repair_segment.go
View File

@ -80,6 +80,7 @@ func cmdRepairSegment(cmd *cobra.Command, args []string) (err error) {
db.RepairQueue(),
db.Buckets(),
db.OverlayCache(),
db.NodeEvents(),
db.Reputation(),
db.Containment(),
rollupsWriteCache,

1
cmd/satellite/repairer.go
View File

@ -69,6 +69,7 @@ func cmdRepairerRun(cmd *cobra.Command, args []string) (err error) {
db.RepairQueue(),
db.Buckets(),
db.OverlayCache(),
db.NodeEvents(),
db.Reputation(),
db.Containment(),
rollupsWriteCache,

2
cmd/tools/segment-verify/main.go
View File

@ -149,7 +149,7 @@ func verifySegmentsRange(cmd *cobra.Command, args []string) error {
dialer := rpc.NewDefaultDialer(tlsOptions)
// setup dependencies for verification
overlay, err := overlay.NewService(log.Named("overlay"), db.OverlayCache(), nil, "", "", satelliteCfg.Overlay)
overlay, err := overlay.NewService(log.Named("overlay"), db.OverlayCache(), db.NodeEvents(), nil, "", "", satelliteCfg.Overlay)
if err != nil {
return Error.Wrap(err)
}

2
private/testplanet/satellite.go
View File

@ -673,7 +673,7 @@ func (planet *Planet) newRepairer(ctx context.Context, index int, identity *iden
rollupsWriteCache := orders.NewRollupsWriteCache(log.Named("orders-write-cache"), db.Orders(), config.Orders.FlushBatchSize)
planet.databases = append(planet.databases, rollupsWriteCacheCloser{rollupsWriteCache})
return satellite.NewRepairer(log, identity, metabaseDB, revocationDB, db.RepairQueue(), db.Buckets(), db.OverlayCache(), db.Reputation(), db.Containment(), rollupsWriteCache, versionInfo, &config, nil)
return satellite.NewRepairer(log, identity, metabaseDB, revocationDB, db.RepairQueue(), db.Buckets(), db.OverlayCache(), db.NodeEvents(), db.Reputation(), db.Containment(), rollupsWriteCache, versionInfo, &config, nil)
}
type rollupsWriteCacheCloser struct {

2
satellite/api.go
View File

@ -291,7 +291,7 @@ func NewAPI(log *zap.Logger, full *identity.FullIdentity, db DB,
{ // setup overlay
peer.Overlay.DB = peer.DB.OverlayCache()
peer.Overlay.Service, err = overlay.NewService(peer.Log.Named("overlay"), peer.Overlay.DB, peer.Mail.Service, config.Console.ExternalAddress, config.Console.SatelliteName, config.Overlay)
peer.Overlay.Service, err = overlay.NewService(peer.Log.Named("overlay"), peer.Overlay.DB, peer.DB.NodeEvents(), peer.Mail.Service, config.Console.ExternalAddress, config.Console.SatelliteName, config.Overlay)
if err != nil {
return nil, errs.Combine(err, peer.Close())
}

2
satellite/core.go
View File

@ -252,7 +252,7 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB,
{ // setup overlay
peer.Overlay.DB = peer.DB.OverlayCache()
peer.Overlay.Service, err = overlay.NewService(peer.Log.Named("overlay"), peer.Overlay.DB, peer.Mail.Service, config.Console.ExternalAddress, config.Console.SatelliteName, config.Overlay)
peer.Overlay.Service, err = overlay.NewService(peer.Log.Named("overlay"), peer.Overlay.DB, peer.DB.NodeEvents(), peer.Mail.Service, config.Console.ExternalAddress, config.Console.SatelliteName, config.Overlay)
if err != nil {
return nil, errs.Combine(err, peer.Close())
}

2
satellite/overlay/benchmark_test.go
View File

@ -359,7 +359,7 @@ func BenchmarkNodeSelection(b *testing.B) {
}
})
service, err := overlay.NewService(zap.NewNop(), overlaydb, nil, "", "", overlay.Config{
service, err := overlay.NewService(zap.NewNop(), overlaydb, db.NodeEvents(), nil, "", "", overlay.Config{
Node: nodeSelectionConfig,
NodeSelectionCache: overlay.UploadSelectionCacheConfig{
Staleness: time.Hour,

5
satellite/overlay/service.go
View File

@ -20,6 +20,7 @@ import (
"storj.io/storj/satellite/geoip"
"storj.io/storj/satellite/mailservice"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeevents"
)
// ErrEmptyNode is returned when the nodeID is empty.
@ -293,6 +294,7 @@ func (node *SelectedNode) Clone() *SelectedNode {
type Service struct {
log *zap.Logger
db DB
nodeEvents nodeevents.DB
mail *mailservice.Service
satelliteName string
satelliteAddress string
@ -304,7 +306,7 @@ type Service struct {
}
// NewService returns a new Service.
func NewService(log *zap.Logger, db DB, mailService *mailservice.Service, satelliteAddr, satelliteName string, config Config) (*Service, error) {
func NewService(log *zap.Logger, db DB, nodeEvents nodeevents.DB, mailService *mailservice.Service, satelliteAddr, satelliteName string, config Config) (*Service, error) {
err := config.Node.AsOfSystemTime.isValid()
if err != nil {
return nil, errs.Wrap(err)
@ -336,6 +338,7 @@ func NewService(log *zap.Logger, db DB, mailService *mailservice.Service, satell
return &Service{
log: log,
db: db,
nodeEvents: nodeEvents,
mail: mailService,
satelliteAddress: satelliteAddr,
satelliteName: satelliteName,

7
satellite/overlay/service_test.go
View File

@ -25,6 +25,7 @@ import (
"storj.io/storj/private/testplanet"
"storj.io/storj/satellite"
"storj.io/storj/satellite/console/consoleweb/consoleapi"
"storj.io/storj/satellite/nodeevents"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/reputation"
"storj.io/storj/satellite/satellitedb/satellitedbtest"
@ -34,7 +35,7 @@ func TestCache_Database(t *testing.T) {
t.Parallel()
satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) {
testCache(ctx, t, db.OverlayCache())
testCache(ctx, t, db.OverlayCache(), db.NodeEvents())
})
}
@ -57,7 +58,7 @@ func testAuditHistoryConfig() reputation.AuditHistoryConfig {
}
}
func testCache(ctx *testcontext.Context, t *testing.T, store overlay.DB) {
func testCache(ctx *testcontext.Context, t *testing.T, store overlay.DB, nodeEvents nodeevents.DB) {
valid1ID := testrand.NodeID()
valid2ID := testrand.NodeID()
valid3ID := testrand.NodeID()
@ -76,7 +77,7 @@ func testCache(ctx *testcontext.Context, t *testing.T, store overlay.DB) {
serviceCtx, serviceCancel := context.WithCancel(ctx)
defer serviceCancel()
service, err := overlay.NewService(zaptest.NewLogger(t), store, nil, "", "", serviceConfig)
service, err := overlay.NewService(zaptest.NewLogger(t), store, nodeEvents, nil, "", "", serviceConfig)
require.NoError(t, err)
ctx.Go(func() error { return service.Run(serviceCtx) })
defer ctx.Check(service.Close)

4
satellite/repair/checker/online_test.go
View File

@ -16,6 +16,7 @@ import (
"storj.io/common/testcontext"
"storj.io/common/testrand"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeevents"
"storj.io/storj/satellite/overlay"
)
@ -23,7 +24,7 @@ func TestReliabilityCache_Concurrent(t *testing.T) {
ctx := testcontext.New(t)
defer ctx.Cleanup()
overlayCache, err := overlay.NewService(zap.NewNop(), fakeOverlayDB{}, nil, "", "", overlay.Config{
overlayCache, err := overlay.NewService(zap.NewNop(), fakeOverlayDB{}, fakeNodeEvents{}, nil, "", "", overlay.Config{
NodeSelectionCache: overlay.UploadSelectionCacheConfig{
Staleness: 2 * time.Nanosecond,
},
@ -52,6 +53,7 @@ func TestReliabilityCache_Concurrent(t *testing.T) {
}
type fakeOverlayDB struct{ overlay.DB }
type fakeNodeEvents struct{ nodeevents.DB }
func (fakeOverlayDB) Reliable(context.Context, *overlay.NodeCriteria) (storj.NodeIDList, error) {
return storj.NodeIDList{

4
satellite/repairer.go
View File

@ -28,6 +28,7 @@ import (
"storj.io/storj/satellite/buckets"
"storj.io/storj/satellite/mailservice"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeevents"
"storj.io/storj/satellite/orders"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/repair/queue"
@ -82,6 +83,7 @@ func NewRepairer(log *zap.Logger, full *identity.FullIdentity,
repairQueue queue.RepairQueue,
bucketsDB buckets.DB,
overlayCache overlay.DB,
nodeEvents nodeevents.DB,
reputationdb reputation.DB,
containmentDB audit.Containment,
rollupsWriteCache *orders.RollupsWriteCache,
@ -156,7 +158,7 @@ func NewRepairer(log *zap.Logger, full *identity.FullIdentity,
{ // setup overlay
var err error
peer.Overlay, err = overlay.NewService(log.Named("overlay"), overlayCache, peer.Mail, config.Console.ExternalAddress, config.Console.SatelliteName, config.Overlay)
peer.Overlay, err = overlay.NewService(log.Named("overlay"), overlayCache, nodeEvents, peer.Mail, config.Console.ExternalAddress, config.Console.SatelliteName, config.Overlay)
if err != nil {
return nil, errs.Combine(err, peer.Close())
}