storj/satellite/overlay/uploadselection.go

// Copyright (C) 2019 Storj Labs, Incache.
// See LICENSE for copying information.

package overlay

import (
	"context"
	"time"

	"go.uber.org/zap"

	"storj.io/common/sync2"
	"storj.io/storj/satellite/nodeselection"
)

// UploadSelectionDB implements the database for upload selection cache.
//
// architecture: Database
type UploadSelectionDB interface {
	// SelectAllStorageNodesUpload returns all nodes that qualify to store data, organized as reputable nodes and new nodes
	SelectAllStorageNodesUpload(ctx context.Context, selectionCfg NodeSelectionConfig) (reputable, new []*nodeselection.SelectedNode, err error)
}

// UploadSelectionCacheConfig is a configuration for upload selection cache.
type UploadSelectionCacheConfig struct {
	Disabled  bool          `help:"disable node cache" default:"false" deprecated:"true"`
	Staleness time.Duration `help:"how stale the node selection cache can be" releaseDefault:"3m" devDefault:"5m" testDefault:"3m"`
}

// UploadSelectionCache keeps a list of all the storage nodes that are qualified to store data
// We organize the nodes by if they are reputable or a new node on the network.
// The cache will sync with the nodes table in the database and get refreshed once the staleness time has past.
type UploadSelectionCache struct {
	log             *zap.Logger
	db              UploadSelectionDB
	selectionConfig NodeSelectionConfig

	cache sync2.ReadCacheOf[*nodeselection.State]

	defaultFilters nodeselection.NodeFilters
	placementRules PlacementRules
}

// NewUploadSelectionCache creates a new cache that keeps a list of all the storage nodes that are qualified to store data.
func NewUploadSelectionCache(log *zap.Logger, db UploadSelectionDB, staleness time.Duration, config NodeSelectionConfig, defaultFilter nodeselection.NodeFilters, placementRules PlacementRules) (*UploadSelectionCache, error) {
	cache := &UploadSelectionCache{
		log:             log,
		db:              db,
		selectionConfig: config,
		defaultFilters:  defaultFilter,
		placementRules:  placementRules,
	}
	return cache, cache.cache.Init(staleness/2, staleness, cache.read)
}

// Run runs the background task for cache.
func (cache *UploadSelectionCache) Run(ctx context.Context) (err error) {
	return cache.cache.Run(ctx)
}

// Refresh populates the cache with all of the reputableNodes and newNode nodes
// This method is useful for tests.
func (cache *UploadSelectionCache) Refresh(ctx context.Context) (err error) {
	defer mon.Task()(&ctx)(&err)
	_, err = cache.cache.RefreshAndGet(ctx, time.Now())
	return err
}

// refresh calls out to the database and refreshes the cache with the most up-to-date
// data from the nodes table, then sets time that the last refresh occurred so we know when
// to refresh again in the future.
func (cache *UploadSelectionCache) read(ctx context.Context) (_ *nodeselection.State, err error) {
	defer mon.Task()(&ctx)(&err)

	reputableNodes, newNodes, err := cache.db.SelectAllStorageNodesUpload(ctx, cache.selectionConfig)
	if err != nil {
		return nil, Error.Wrap(err)
	}

	state := nodeselection.NewState(reputableNodes, newNodes)

	mon.IntVal("refresh_cache_size_reputable").Observe(int64(len(reputableNodes)))
	mon.IntVal("refresh_cache_size_new").Observe(int64(len(newNodes)))

	return state, nil
}

// GetNodes selects nodes from the cache that will be used to upload a file.
// Every node selected will be from a distinct network.
// If the cache hasn't been refreshed recently it will do so first.
func (cache *UploadSelectionCache) GetNodes(ctx context.Context, req FindStorageNodesRequest) (_ []*nodeselection.SelectedNode, err error) {
	defer mon.Task()(&ctx)(&err)

	state, err := cache.cache.Get(ctx, time.Now())
	if err != nil {
		return nil, Error.Wrap(err)
	}

	placementRules := cache.placementRules(req.Placement)
	useSubnetExclusion := !nodeselection.AllowSameSubnet(placementRules)

	filters := nodeselection.NodeFilters{placementRules}
	if len(req.ExcludedIDs) > 0 {
		if useSubnetExclusion {
			filters = append(filters, state.ExcludeNetworksBasedOnNodes(req.ExcludedIDs))
		} else {
			filters = append(filters, nodeselection.ExcludedIDs(req.ExcludedIDs))
		}
	}

	filters = append(filters, cache.defaultFilters)

	selectionReq := nodeselection.Request{
		Count:       req.RequestedCount,
		NewFraction: cache.selectionConfig.NewNodeFraction,
		NodeFilters: filters,
	}

	if !useSubnetExclusion {
		selectionReq.SelectionType = nodeselection.SelectionTypeByID
	}

	selected, err := state.Select(ctx, selectionReq)
	if nodeselection.ErrNotEnoughNodes.Has(err) {
		err = ErrNotEnoughNodes.Wrap(err)
	}
	return selected, err
}